package nl.altenpts.net.kad;

import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.security.KeyFactory;
import java.security.PublicKey;
import java.security.spec.InvalidKeySpecException;
import java.security.spec.X509EncodedKeySpec;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;

import nl.altenpts.net.broker.IncomingMessage;
import nl.altenpts.net.broker.IncomingRequest;
import nl.altenpts.net.broker.OutgoingMessage;
import nl.altenpts.net.broker.RequestBroker;
import nl.altenpts.net.broker.RequestHandler;
import nl.altenpts.net.broker.ResponseHandler;
import nl.altenpts.net.broker.SignedMessageBroker;
import nl.altenpts.net.broker.SignedMessageHook;

/**
 * 
 * Kademlia broker provides methods to send Kademlia requests. KBroker also handles
 * all incoming kademlia requests  
 * 
 * @author fabian
 *
 */
public class KBroker {
	
	/**
	 * Construct a new KBroker
	 * 
	 * @param broker The request broker which is used to send requests and respond to requests 
	 * @param table Kademlia routing table 
	 * @param store Value store
	 */
	public KBroker(RequestBroker broker, RoutingTable table, ValueStore store) {
		this.broker = broker;
		this.table = table;
		this.store = store;
		
		this.broker.setHandler(new KadRequestHandler());
		
		SignedMessageBroker msgbroker = broker.getBroker();
		msgbroker.addHook(new IncomingMessageHook());
	}
	
	
	/**
	 * 
	 * 
	 * @param host
	 * @param port
	 * @param handler
	 */
	public void PingRequest(InetSocketAddress address, ResponseHandler handler) {
		//System.out.println("Ping request to " + port);
		OutgoingMessage message = new OutgoingMessage(address);
		ByteBuffer buffer = ByteBuffer.allocate(1);
		buffer.put(KADBROKER_PING_REQUEST);
		buffer.flip();
		message.prependContent(buffer);
		broker.sendRequest(message, handler, KADBROKER_TIMEOUT);
	}
	
	
	/**
	 * Send a ping to Node n
	 * 	
	 * @param n target node
	 * @param handler response handler
	 */
	public void PingRequest(Node n, ResponseHandler handler) {
		PingRequest(n.getAddress(), handler);	
	}
	
	
	/**
	 * Request the closest nodes to id from Node n
	 * 
	 * @param address
	 * @param id
	 * @param handler
	 */
	public void NodeRequest(InetSocketAddress address, Id id, ResponseHandler handler) {
		OutgoingMessage message = new OutgoingMessage(address);
		ByteBuffer buffer = ByteBuffer.allocate(5 + id.getByteArray().length);
		buffer.put(KADBROKER_NODE_REQUEST);
		buffer.putInt(id.getByteArray().length);
		buffer.put(id.getByteArray());
		buffer.flip();
		message.prependContent(buffer);
		broker.sendRequest(message, handler, KADBROKER_TIMEOUT);
	}
	
	
	/**
	 * Request the closest nodes to id from Node n
	 * 
	 * @param n
	 * @param id
	 * @param handler
	 */
	public void NodeRequest(Node n, Id id, ResponseHandler handler) {
		NodeRequest(n.getAddress(), id, handler);
	}
	
	public void ValueRequest(Node n, byte[] key, ResponseHandler handler) {
		//System.out.println("Value request to " + n.getPort());
		OutgoingMessage message = new OutgoingMessage(n.getAddress());
		ByteBuffer buffer = ByteBuffer.allocate(5 + key.length);
		buffer.put(KADBROKER_VALUE_REQUEST);
		buffer.putInt(key.length);
		buffer.put(key);
		buffer.flip();
		message.prependContent(buffer);
		broker.sendRequest(message, handler, KADBROKER_TIMEOUT);
	}
	
	public void StoreRequest(Node n, Value value, ResponseHandler handler) {
		//System.out.println("Store request to " + n.getPort());
		byte[] bkey = value.getKey();
		byte[] bvalue = value.getValue();
		byte[] bsetterkey = value.getSetter().getEncoded();
		byte[] bsign = value.getSign();
		
		OutgoingMessage message = new OutgoingMessage(n.getAddress());
		
		ByteBuffer buffer = ByteBuffer.allocate(17 + bkey.length + bvalue.length+
				bsetterkey.length + bsign.length);
		
		buffer.put(KADBROKER_STORE_REQUEST);
		buffer.putInt(bkey.length);
		buffer.put(bkey);
		buffer.putInt(bvalue.length);
		buffer.put(bvalue);
		buffer.putInt(bsetterkey.length);
		buffer.put(bsetterkey);
		buffer.putInt(bsign.length);
		buffer.put(bsign);
				
		buffer.flip();
		message.prependContent(buffer);
		broker.sendRequest(message, handler, KADBROKER_TIMEOUT);
	}
	
	static List<Node> parseNodeList(ByteBuffer buffer) {
		List<Node> nodes = new LinkedList<Node>();
		
		int numnodes = buffer.getInt();
		for(int i = 0; i < numnodes; i++) {
			try {
				nodes.add(new Node(buffer));
			} catch (Exception e) {
				// ignore node if it could not be parsed
			}
		}		
		return nodes;		
	}
	
	static Set<Value> parseValueList(ByteBuffer buffer) throws InvalidKeySpecException {
		Set<Value> values = new HashSet<Value>();
		
		int numnodes = buffer.getInt();
		for(int i = 0; i < numnodes; i++) {
			values.add(new Value(buffer));
		}
		return values;
	}

	static void writeNodes(OutgoingMessage message, List<Node> nodes) {
		Iterator<Node> iterator = nodes.iterator();
		ByteBuffer buffer = ByteBuffer.allocate(4);
		
		buffer.putInt(nodes.size());
		buffer.flip();
		message.appendContent(buffer);
		
		while(iterator.hasNext()) {
			buffer = iterator.next().toByteBuffer();
		    message.appendContent(buffer);
		}	
	}
	
	static void writeValues(OutgoingMessage message, Set<Value> values) {
		Iterator<Value> iterator = values.iterator();
		ByteBuffer buffer = ByteBuffer.allocate(4);
		
		buffer.putInt(values.size());
		buffer.flip();
		message.appendContent(buffer);
		
		while(iterator.hasNext()) {
			buffer = iterator.next().toByteBuffer();
			message.appendContent(buffer);
		}	
	}
	

	/**
	 * Request handler for incoming Kademlia requests.
	 * 
	 * @author fabian
	 *
	 */
	private class KadRequestHandler implements RequestHandler {

		public void onIncoming(IncomingRequest request, PublicKey publickey) {
			ByteBuffer buffer = request.getMessage().getPayload();
			byte requesttype = buffer.get();
			Id target = null;
			ByteBuffer bufout = ByteBuffer.allocate(1);
			
			OutgoingMessage message = new OutgoingMessage(request.getMessage().getAddress());
			
			switch(requesttype) {
			case KADBROKER_STORE_REQUEST:
				
				try {
					byte[] key = new byte[buffer.getInt()];
					buffer.get(key);

					byte[] value = new byte[buffer.getInt()];
					buffer.get(value);

					byte[] bsetterkey = new byte[buffer.getInt()];
					buffer.get(bsetterkey);				

					byte[] sign = new byte[buffer.getInt()];
					buffer.get(sign);

					X509EncodedKeySpec keyspec = new X509EncodedKeySpec(bsetterkey);
					KeyFactory keyfactory = KeyFactory.getInstance(Kademlia.keyalg);
					PublicKey setterkey = keyfactory.generatePublic(keyspec);

					Value v = new Value(key, value, setterkey, sign);				
					store.put(v);
					System.out.println(table.getRoot().getAddress().getPort() + ": storereq from " +
							message.getAddress().getPort());
				} catch (Exception ex) {
					//store failed, don't send response
					break;					
				}
				// intentionally no break here	
			case KADBROKER_PING_REQUEST:
				if(requesttype == KADBROKER_PING_REQUEST) {
					System.out.println(table.getRoot().getAddress().getPort() + ": pingrequest from " +
							message.getAddress().getPort());
				}
				// send empty response
				request.respond(message);
				break;
			case KADBROKER_VALUE_REQUEST:
				byte[] key = new byte[buffer.getInt()];
				buffer.get(key);
				
				System.out.println(table.getRoot().getAddress().getPort() + ": valuereq from " +
						message.getAddress().getPort());
				
				Set<Value> values = store.get(key);					
				if(values != null && !values.isEmpty()) {
					StringBuffer strvalues = new StringBuffer();
					for(Iterator<Value> iterator = values.iterator(); iterator.hasNext(); ) {
						Value v = iterator.next();
						strvalues.append(Arrays.toString(v.getValue()));
						strvalues.append('-');
					}
					System.out.println(table.getRoot().getAddress().getPort() + ": sending values " + strvalues.toString());
					bufout.put(KADBROKER_VALUE_RESPONSE);
					bufout.flip();
					message.appendContent(bufout);
					writeValues(message, values);
					request.respond(message);
					break;
				} else {
					System.out.println("Sending nodes");
					target = Kademlia.keyToId(key);
					bufout.put(KADBROKER_NODE_RESPONSE);				
					bufout.flip();
					message.appendContent(bufout);
				}
				//intentionally no break here
			case KADBROKER_NODE_REQUEST:
				
				if(target == null) {
					byte[] idbytes = new byte[buffer.getInt()];
					buffer.get(idbytes);
					target = new Id(idbytes);
					System.out.println(table.getRoot().getAddress().getPort() + ": nodereq from " +
							message.getAddress().getPort());
				}	
				
				List<Node> nodes = table.findNodes(target, Kademlia.k);
				
				StringBuffer strvalues = new StringBuffer();
				for(Iterator<Node> iterator = nodes.iterator(); iterator.hasNext(); ) {
					Node n = iterator.next();
					strvalues.append(n.getAddress().getPort());
					strvalues.append('-');
				}
				System.out.println(table.getRoot().getAddress().getPort() + ": sending nodes " + strvalues.toString());
				
				writeNodes(message, nodes);
				request.respond(message);
				break;
			}
		}
	}
	
	private class IncomingMessageHook implements SignedMessageHook {

		public void onIncoming(IncomingMessage message, PublicKey key) {
			Node n = new Node(key, message.getAddress());
			n.updateLastSeen();
			table.updateNode(n, KBroker.this);			
		}		
	}
	
	private RequestBroker broker;
	private RoutingTable table;
	private ValueStore store;
	
	final static byte KADBROKER_PING_REQUEST   = 1;
	final static byte KADBROKER_NODE_REQUEST   = 2;
	final static byte KADBROKER_VALUE_REQUEST  = 3;
	final static byte KADBROKER_STORE_REQUEST  = 4;
	
	final static byte KADBROKER_NODE_RESPONSE  = 2;
	final static byte KADBROKER_VALUE_RESPONSE = 3;
	
	final static int KADBROKER_TIMEOUT = 30000;

}
